package com.fangsheng.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.common.protocol.body.ClusterInfo;
import org.apache.rocketmq.common.protocol.body.TopicList;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.common.protocol.route.BrokerData;
import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
import org.junit.jupiter.api.Test;

import java.io.IOException;
import java.util.*;

/**
 * @创建人: 放生
 * @创建时间: 2022/1/20
 * @描述: rocketmq 集群的方式案列， 并且有消息消费失败的处理逻辑，消息顺序消费 ，广播消费
 */
public class ExampleRocketMQ {





    @Test
    public void admin() throws Exception {

        DefaultMQAdminExt admin = new DefaultMQAdminExt();
        // 一般都是三台，
        admin.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876;192.168.10.31:9876");
        admin.start();

        ClusterInfo clusterInfo = admin.examineBrokerClusterInfo();
        HashMap<String, BrokerData> brokerAddrTable = clusterInfo.getBrokerAddrTable();
        Set<Map.Entry<String, BrokerData>> entries = brokerAddrTable.entrySet();
        Iterator<Map.Entry<String, BrokerData>> iter = entries.iterator();
        while(iter.hasNext()){
            Map.Entry<String, BrokerData> next = iter.next();
            System.out.println(next.getKey()+ " "+ next.getValue());
        }
        TopicList topicList = admin.fetchAllTopicList();
        topicList.getTopicList().forEach(s->{
            System.out.println(s);
        });
    }

    //--------------example 001  start(消息异常消费  重试)------------------------
    @Test
    public void producer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ooxx");
        producer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");
        producer.start();

        ArrayList<Message> msgs = new ArrayList<>();

        for (int i = 0; i < 10; i++) {
            msgs.add(new Message(
                    "bala",
                    "TagA",
                    "key"+i,//全局唯一
                    ("message:"+i).getBytes()

            ));
        }

        SendResult res = producer.send(msgs);
        System.out.println(res);
    }


    /**
     * messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
     * 18个延迟级别
     * 1：
     * ConsumeConcurrentlyStatus.RECONSUME_LATER
     * consumer.setMaxReconsumeTimes(2); 设置重复消费的次数
     * %RETRY%xxx3x （broker 默认维护重试队列的格式）
     * %DLQ%xxx3x （broker 默认维护死信队列的格式）
     * 总结：rocketmq：重试队列，死信队列：以topic形式，且：以consumerGroup为单位，为每一个消费者组维护一个重试队列，和死信队列
     * 补充一下:kafka，客户端自己维护逻辑
     *
     * 2:延迟队列的场景
     */
    @Test
    public void consumer()  {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xxx6x");
        consumer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMaxReconsumeTimes(2);// 设置重复消费的次数，下面的代码中有int a =  1/0;，会报错
        //这里设置了 setMaxReconsumeTimes(2) 表示会重复两次的消费
        //TODO
//        consumer.setConsumeMessageBatchMaxSize(1);   //交付给消费逻辑的数量 就是决定了List<MessageExt> msgs 大小
        consumer.setPullBatchSize(1);  //从broker拉取消息的数量
        try {
            consumer.subscribe("bala","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                MessageExt msge = msgs.get(0);
                try {
                    System.out.println(new String(msge.getBody()));
                    if(msge.getKeys().equals("key1")){
                        int a =  1/0;
                    }
                }catch (Exception e ){
                    // 表示有异常之后在次消费 如果没有设置重复消费的次数，则会走默认的18个级别即messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                // 类似于消费成功然后做的 消费成功的应答
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        /*
        new MessageListenerOrderly()  //有序消息处理
        new MessageListenerConcurrently()
        消息顺序的依赖性

         */

        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        try {
            System.in.read();
        } catch (IOException e) {
            e.printStackTrace();
        }
        consumer.shutdown();

    }

// -------------------------------example 001 end-----------------------------------------------------------------


    //--------------example 002     消息顺序的案列      ------------------------

    /**
     * 保证有序性：
     * 1，producer，其实就是自定义分区器(队列选择器)
     * 2，consumer，MessageListenerOrderly，
     * 感知：
     * 才能实现，生产的时候，有顺序依赖的msg进入一个队列，并且，消费者也会再多线程情况下保证单线程对应的queue能按顺序消费
     * 如果，你只创建一个queue，全局有序
     * @throws Exception
     */
    @Test
    public void orderProducer() throws Exception {
        DefaultMQProducer producer = new DefaultMQProducer("ooxx");
        producer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");
        producer.start();
        for (int i = 0; i < 10; i++) {

            Message msg = new Message(
                    "order_topic",
                    "TagA",
                    ("message body : " + i + "type: " + i % 3).getBytes()
            );
            // 待完善的地方 new MessageQueueSelector() 不能每次都new一个，所以需要全局的一个，我们们也可以自顶一个
            SendResult res = producer.send(msg, new MessageQueueSelector() {   //有没有性能问题？
                @Override
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) { //arg 就是传进来的 i%3
                    Integer index = (Integer) arg;
                    //根据 传进来的 arg 模上当前topic的所有队列的size 然后决定把消息发到哪一个队列中去
                    int target = index % mqs.size();
                    return mqs.get(target);
//                    MessageQueue target = mqs.get(index);
//                    return target;
//
                }
            }, i % 3);
            System.out.println(res);
        };


    }

    @Test
    public void orderConsumer() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xoxo02");
        consumer.setNamesrvAddr("192.168.150.11:9876;192.168.150.12:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        try {
            consumer.subscribe("order_topic","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                // 如果当前topic只有一个队列 那么消息的消费是全局有序的，如果不是，有多个队列那么会有保证局部有序，就是每个队列的里的消息是有序的，
                // 队列之间就不一定有序了 ，
                msgs.forEach(msge->{
                    System.out.println(Thread.currentThread().getName()+ " : "+new String(msge.getBody()));
                });
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.in.read();
        consumer.shutdown();
    }

//  ------------------------------- 广播 --------------------------------------
    /**
     * 广播消息： 只需要修改消费者端就可以了
     *      添加 consumer.setMessageModel(MessageModel.BROADCASTING);即可，因为非广播下同一个消息只能被同一个组的消费者消费，
     *      加了这个设置以后都能收到
     * @throws Exception
     */
    @Test
    public void broadcast() throws Exception {

        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("xoxo02");
        consumer.setNamesrvAddr("192.168.10.11:9876;192.168.10.21:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setMessageModel(MessageModel.BROADCASTING);
        try {
            consumer.subscribe("order_topic","*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }

        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                // 如果当前topic只有一个队列 那么消息的消费是全局有序的，如果不是，有多个队列那么会有保证局部有序，就是每个队列的里的消息是有序的，
                // 队列之间就不一定有序了 ，
                msgs.forEach(msge->{
                    System.out.println(Thread.currentThread().getName()+ " : "+new String(msge.getBody()));
                });
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });

        consumer.start();
        System.in.read();
        consumer.shutdown();
    }


     /*

        广播：consumer端自己规划：
            //set to broadcast mode
        consumer.setMessageModel(MessageModel.BROADCASTING);

        延迟队列 (超时)
        1，细粒度
            时间轮：【先自行学习一下】，排序成本，分治概念
        2，粗粒度
        messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
            目标topic： ooxx_topic   消息，期望 30s，你的消息是放到  30S——topic，由rocketmq broker

        事务
     */



}
